package org.totoro.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.totoro.config.KafkaConfig;

/**
 * @author YHL
 * @version V1.0
 * @Description: 消息提供者
 * @date 2018-11-06
 */
@RestController
@RequestMapping("/producer/")
public class KafkaProducerController {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @GetMapping("/add")
    public Object add(@RequestParam String text) {

        ListenableFuture send = kafkaTemplate.send(KafkaConfig.topic_name, text);

        send.addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                ex.printStackTrace();
            }

            @Override
            public void onSuccess(Object result) {
                System.out.println("成功 " + result);
            }
        });


        return text;
    }

}
